package com.litchi.listener;

import com.litchi.pools.ChangeEventThreadPoolHolder;
import io.debezium.engine.DebeziumEngine;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.util.Assert;

import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author: Jiangxx
 * @Date: 2023/10/18
 * @Description:
 */
@Slf4j
public class DataChangeExecutor implements InitializingBean, SmartLifecycle {

    private final ThreadPoolExecutor executor = ChangeEventThreadPoolHolder.getInnerExecutor();

    private final DebeziumEngine<?> debeziumEngine;

    public DataChangeExecutor(DebeziumEngine<?> debeziumEngine) {
        this.debeziumEngine = debeziumEngine;
    }

    @Override
    public void start() {
        log.warn(ChangeEventThreadPoolHolder.POOL_NAME + "线程池开始执行 debeziumEngine 实时监听任务!");
        executor.execute(debeziumEngine);
    }

    @SneakyThrows
    @Override
    public void stop() {
        log.warn("debeziumEngine 监听实例关闭!");
        debeziumEngine.close();
        Thread.sleep(2000);
        log.warn(ChangeEventThreadPoolHolder.POOL_NAME + "线程池关闭!");
        executor.shutdown();
    }

    @Override
    public boolean isRunning() {
        return false;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.notNull(debeziumEngine, "DebeZiumEngine 不能为空!");
    }
}
